package com.sf.iisp.redis.listener;

import com.sf.iisp.BindingException;
import com.sf.iisp.bind.biz.BindingBiz;
import com.sf.iisp.bind.dao.IBindingDao;
import com.sf.iisp.bind.domain.BindingRecord;
import com.sf.iisp.bind.domain.ReturnBindingInfo;
import com.sf.iisp.bind.enums.State;
import com.sf.iisp.bind.helper.ResponseMessage;
import com.sf.iisp.kafka.adapter.IKafkaProducerPoolDelegate;
import com.sf.kafka.check.util.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;

import javax.annotation.Resource;

/**
 * 预约绑定号码的 Redis 消息监听器
 * Created by qi on 16-7-6.
 */
public class BindingListener implements MessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(BindingListener.class);

    private RedisTemplate<String, Object> redisTemplate;
    private ListOperations<String, Object> listOps;

    @Resource
    private BindingBiz bindingBiz;

    @Qualifier("producer")
    @Resource
    private IKafkaProducerPoolDelegate delegate;
    @Resource
    private IBindingDao bindingDao;

    @Value("${isSendKafka:true}")
    private String isSend;


    public BindingListener(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.listOps = redisTemplate.opsForList();
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        byte[] body = message.getBody();
        byte[] channel = message.getChannel();

        String topic = redisTemplate.getStringSerializer().deserialize(channel);
        String itemValue = (String) redisTemplate.getValueSerializer().deserialize(body);
        long size = listOps.size(itemValue);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("topic = {}, receive message = {}", topic, itemValue);
            LOGGER.debug("receive message size = {}", size);
        }

        //TODO: 保存数据库，绑定号码。考虑性能。
        for (long i = 0; i < size; i++) {
//            BindingRecord br = (BindingRecord)listOps.index(itemValue, i);
            BindingRecord br = (BindingRecord) listOps.rightPop(itemValue);
            LOGGER.debug("br is exists = {}", br);

            if (br != null) {
                LOGGER.info("get redis item binding record: courierPhone = {}, customerPhone ={}. ", br.getCourierPhone(), br.getCustomerPhone());

                ReturnBindingInfo returnInfo = null;
                try {
                    returnInfo = bindingBiz.convertBindingRecord2ReturnInfo(br);
                    BindingRecord history = bindingDao.findBindRecordByPhone(State.SUCCESS.toString(), br.getCourierPhone(), br.getCustomerPhone());
                    if (null != history) {
                        returnInfo.setState(State.EXISTS.toString());
                        returnInfo.setMsg(State.EXISTS.getMsg());
                        returnInfo.setVirtualNo(history.getVirtualNo());
                        returnInfo.setOriginTransId(br.getTransId());
                        LOGGER.info("reserve binding record be bound:  courierPhone = {}, customerPhone = {} , virtual no ={}", history.getCourierPhone(), history.getCustomerPhone(), history.getVirtualNo());
                    } else {
                        ResponseMessage resp = bindingBiz.bind(br.getCourierPhone(), br.getCustomerPhone(), br);
                        LOGGER.info("reserve binding record :  courierPhone = {}, customerPhone = {} , virtual no ={}", br.getCourierPhone(), br.getCustomerPhone(), resp.getMsg());

                        history = bindingDao.findBindRecordByPhone(State.RESERVED.toString(), br.getCourierPhone(), br.getCustomerPhone());
                        if (null != history) {
                            history.setState(State.SUCCESS.toString());
                            history.setVirtualNo(resp.getMsg());
                            bindingDao.updateSaveBindRecord(history);
                        } else {
                            br.setState(State.SUCCESS.toString());
                            br.setVirtualNo(resp.getMsg());
                            bindingDao.updateSaveBindRecord(br);
                        }

                        returnInfo.setState(State.SUCCESS.toString());
                        returnInfo.setVirtualNo(resp.getMsg());
                    }
                } catch (BindingException e) {
                    br.setState(e.getErrorCode());
                    returnInfo.setState(e.getErrorCode());
                    returnInfo.setMsg(e.getMessage());
                    LOGGER.error(e.getMessage(), e);

                } finally {
                    //返回结果给涅槃
                    String content = JsonUtil.writeValueAsString(returnInfo);
                    LOGGER.info("send reserve SGS message = {}", content);
                    //TODO:测试关闭！！！！
                    if ("true".equals(isSend)) {
                        delegate.sendString(content);
                    }
                }
            }
        }
    }
}
